#ifndef _TASK_SEND_H_
#define _TASK_SEND_H_

#include "../define/define.h"
#include "../trans_server/server.h"
#include "../tcp_server/tcp_server.h"
#include "../trans_server/server_recv.h"
#include "../util/timer.h"
#include "./task_map.h"

class task_send : public tcp_sink
{
public:
	
	task_send()
	{
 		m_server = new tcp_server(this);
		m_log    = new logfile("task_send");
	}

    ~task_send()
	{
		delete m_log;
		delete m_server;
	}

	void init(UDPServer* udp)
	{
	   m_udp = udp;
	}

    bool add_task(ip_node task)
	{
		m_lock.EnterLock();
		//m_log->WriteTrace("add task");
	    std::map<ip_node,uint16>::iterator iter;

		char lls[1024] = {0};
		sprintf(lls,"task:port=%d",task.l_port);
		m_log->WriteTrace(lls);

        iter  =m_taskmap.find(task);
		if(iter != m_taskmap.end())
		{
			m_lock.LeaveLock();
			return false;
		}

		m_server->add_port(task.l_port);
		m_taskmap.insert(std::map<ip_node,uint16>::value_type(task,task.l_port));
	
		m_tasklist.push_back(task);

		m_lock.LeaveLock();
		return true;
	}

	bool del_task(ip_node task)
	{
		m_lock.EnterLock();
		//m_log->WriteTrace("del task");
	    std::map<ip_node,uint16>::iterator iter;

		char lls[1024] = {0};
		sprintf(lls,"del task:port=%d",task.l_port);
		m_log->WriteTrace(lls);

        iter  = m_taskmap.find(task);
		if(iter != m_taskmap.end())
		{
			m_lock.LeaveLock();
			return false;
		}

		m_taskmap.erase(iter);
		
		//m_log->WriteTrace("del task");
		m_server->del_port(task.l_port);		

		vector<ip_node> ::iterator itr = m_tasklist.begin();
		while (itr != m_tasklist.end())
		{
			if(itr->l_port == task.l_port)
			{
				m_tasklist.erase(itr);
				m_lock.LeaveLock();
				return true;
			}	    
			++itr;
		}

		m_lock.LeaveLock();
		return false;
	}

	uint32 get_all_task(ip_node* task,uint32 count)
	{
		m_lock.EnterLock();
		uint32 i =0;
		vector<ip_node> ::iterator itr = m_tasklist.begin();
		while (itr != m_tasklist.end())
		{
			if(itr->d_ip == 0)
				break;

			if(count < i)
				break;

			//fixme
			//memcpy(task + i,&itr,sizeof(ip_node));
	
			++itr;
		
			i++;
		}

		m_lock.LeaveLock();
		return i;
	}

	bool ack_data(ip_node task, uint32 con_id,uint32 socket,uint32 d_socket,char* buf, uint32 buf_size)
	{
		m_lock.EnterLock();

		ip_node_key node_key;
		node_key.d_ip = task.d_ip;
		node_key.d_port = task.d_port;
		node_key.l_port  = task.l_port;
		node_key.s_socket = socket;

        std::map<ip_node_key,uint32>::iterator it;
        
		it= m_map.find(node_key);

		if(it == m_map.end())
		{
			char ll[1023] = {0};
			sprintf(ll,"ack:task && recv map error,socket=%d,d_socket=%d",
				socket,d_socket);
            m_log->WriteTrace(ll);
			m_lock.LeaveLock();
			return false;
		}

		if(it->second != socket)
		{
			char ll[1023] = {0};
			sprintf(ll,"ack:task && recv map error,socket error,socket=d%,d_socket=%d",
				socket,d_socket);
            m_log->WriteTrace(ll);
			m_lock.LeaveLock();
			return true;
		}

		//m_log->WriteTrace("ask data");
		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(socket);
    
		if(iter == m_sendmap.end())
		{
			m_lock.LeaveLock();
			return false;
		}

		iter->second->set_d_socket(d_socket);
		iter->second->ack_data(con_id,socket,buf,buf_size);
	    m_lock.LeaveLock();
		return  true;
	}

	bool add_buf(ip_node task,uint32 my_socket,uint32 d_socket,char* buf, uint32 buf_size,uint32 data_id)
	{
		m_lock.EnterLock();

		ip_node_key node_key;
		node_key.d_ip = task.d_ip;
		node_key.d_port = task.d_port;
		node_key.l_port  = task.l_port;
		node_key.s_socket = my_socket;
		
		/*char ll[1024] = {0};
		in_addr ip; 
		memcpy(&ip,&task.d_ip,sizeof(uint32));
		sprintf(ll,"m_map.size=%d,socket=%d,d_socket=%d,node_key.d_port=%d,task.l_port=%d,d_ip=%s",
			m_map.size(),my_socket,d_socket,task.d_port,task.l_port,inet_ntoa(ip));
		m_log->WriteTrace(ll);*/	
		
		std::map<ip_node_key,uint32>::iterator it;
        
		it= m_map.find(node_key);
		if(it == m_map.end())
		{
			char ll[1024] = {0};
			sprintf(ll,"add_buf:task && recv map error,socket=%d,d_socket=%d",
				my_socket,d_socket);
			m_log->WriteTrace(ll);
			m_lock.LeaveLock();
			return false;
		}

		if(it->second != my_socket)
		{
			char ll[1024] = {0};
			sprintf(ll,"add_buf:task && recv map error,socket error,socket=%d,d_socket=%d",
				my_socket,d_socket);
			m_log->WriteTrace(ll);
			m_lock.LeaveLock();
			return true;
		}

		std::map<uint32,server*>::iterator iter2;
		iter2 = m_sendmap.find(my_socket);

		if(iter2 != m_sendmap.end())
		     iter2->second->set_d_socket(d_socket);

		std::map<uint32,server_recv*>::iterator iter1;
		iter1 = m_recvmap.find(my_socket);

		if(iter1 != m_recvmap.end())
		{
			uint32 rt = iter1->second->put_data(buf,buf_size,data_id);

			
			if(rt == 2)
				OnError(my_socket);

			m_lock.LeaveLock();
			return true;
		}

		m_lock.LeaveLock();
	}

	bool OnAccept(uint32 socket, sockaddr sock_addr,uint16 port)
	{
		m_lock.EnterLock();
     
		std::map<ip_node,uint16>::iterator it;
		it = m_taskmap.begin();

		while(it != m_taskmap.end())
		{
			if(it->first.l_port == port)
				break;
			
			it++;
		}

		if(it == m_taskmap.end())
		{
			m_log->WriteTrace("no this port");
			m_lock.LeaveLock();
			return false;
		}

		m_server->add_socket(it->first.l_port,socket);
		
		ip_node_key node_key ;
		node_key.d_ip = it->first.d_ip;
		node_key.d_port = it->first.d_port;
		node_key.l_port  = it->first.l_port;
        node_key.s_socket = socket;

		std::map<ip_node_key,uint32>::iterator map_iter;
		map_iter = m_map.find(node_key);
		
   		m_map.insert(std::map<ip_node_key,uint32>::value_type(node_key,socket));
        
		/*char ll[1024] = {0};
		in_addr ip; 
		memcpy(&ip,&it->first.d_ip,sizeof(uint32));
		sprintf(ll,"add m_map : m_map.size=%d,socket=%d,node_key.d_port=%d,task.l_port=%d,d_ip=%s",
			m_map.size(),socket,it->first.d_port,it->first.l_port,inet_ntoa(ip));
		m_log->WriteTrace(ll);*/

		ip_node     node = it->first;
		server*     send = m_sendlist.get_send(m_udp,socket,0,DATA_send);
		
		send->add_node(node);
		
		server_recv* s_recv = m_recvlist.get_recv(socket);

		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(socket);

		if(iter != m_sendmap.end())
		{
			m_lock.LeaveLock();
			return false;
		}
        
		m_sendmap.insert(std::map<uint32,server*>::value_type(socket,send));

		std::map<uint32,server_recv*>::iterator iter1;
		iter1 = m_recvmap.find(socket);

		if(iter1 != m_recvmap.end())
		{
			m_lock.LeaveLock();
			return false;
		}
		
		m_recvmap.insert(std::map<uint32,server_recv*>::value_type(socket,s_recv));

		m_lock.LeaveLock();
		return true;
	}

	bool OnRead(uint32 socket, char* buf, uint32 buf_size)
	{
		m_lock.EnterLock();
		
		std::map<ip_node_key,uint32>::iterator it;
		for(it = m_map.begin(); it != m_map.end(); it++)
		{
			if(it->second == socket)
    		    break;
		}

		if(it->second != socket)
		{
			m_log->WriteTrace("read error not find task");
			m_lock.LeaveLock();
			return true;
		}

		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(socket);

		if(iter == m_sendmap.end())
		{
			m_lock.LeaveLock();
			return false;
		}
        
		int i;
		for(i=0;i< buf_size / 1024 ; i++)
			iter->second->Put_data(buf + i* 1024,1024);

		if(buf_size % 1024 > 0)
			iter->second->Put_data(buf + i* 1024,buf_size - i*1024);

        m_lock.LeaveLock();
		return true;
	}

	bool OnSend()
	{
		uint32 socket_list[128] = {0};
		uint32 count =0;

		std::map<uint32,server_recv*>::iterator iter;
		m_lock.EnterLock();

		iter = m_recvmap.begin();
		while(iter != m_recvmap.end())
		{
			char   buf[1024] = {0};
		    uint32 buf_size = 1024;

			uint32 data_id = 0xffffffff;
			uint32 rt = iter->second->get_data(buf,&buf_size,&data_id);

			if(m_recvmap.size() > 0 )
			{
				/*char lls[1024] = {0};
				sprintf(lls,"rt=%d,buf_size=%d,data_id=%d,size=%d,socket=%d,time=%d",
					rt,buf_size,data_id,m_recvmap.size(),iter->first,get_time());
				m_log->WriteTrace(lls);*/
			}

		    if(rt == 0 || rt == 3 || buf_size == 0)
			{
				iter++;
				continue;
			}
			
			if(rt == 2)
			{
				m_log->WriteTrace("get_data down over");
				socket_list[count] = iter->first;
				count++;
				iter++;
				continue;
			}
								
			int size = m_server->sendbuf(iter->first,buf,buf_size); 

			if(size == -1)
			{
				 m_log->WriteTrace("socket error, close");
				 socket_list[count] = iter->first;
				 count++;
				 iter++;
				 continue;
			}

			if(size == 0)
			{
				iter++;
				continue;
			}

			int rt_size = iter->second->set_size(data_id,size);

			if(rt_size == 2)
			{
				m_log->WriteTrace("set_size down over");
				socket_list[count] = iter->first;
				count++;
			}
			iter++;
		}

		m_lock.LeaveLock();

		for(uint32 j=0;j<count;j++)
			OnError(socket_list[j]);
		
		return true;
	}

	bool OnError(uint32 socket)
	{
		char lls[1024] = {0};
		sprintf(lls,"error :socket=%d,time=%d",socket,get_time());
		m_log->WriteTrace(lls);

		std::map<ip_node_key,uint32>::iterator it;
		for(it = m_map.begin(); it!=m_map.end(); it++)
		{
			if(it->second == socket)
			    break;
		}

		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(socket);

		if(iter == m_sendmap.end())
		{
			m_log->WriteTrace("clear server error");
			m_lock.LeaveLock();
			return false;
		}
        
		iter->second->send_down_over();

		m_lock.LeaveLock();

		clear(socket);		
		return true;
	}

	bool clear(uint32 socket)
	{
		m_lock.EnterLock();

		std::map<ip_node_key,uint32>::iterator it;
		for(it = m_map.begin(); it!=m_map.end(); it++)
		{
			if(it->second == socket)
			{
				char ll[1024] = {0};
				sprintf(ll,"delete m_map iter,m_map.size=%d,socket=%d,iter->socket=%d",
					m_map.size(),socket,it->second);
				m_log->WriteTrace(ll);
               	m_map.erase(it);
			    break;
			}
		}

		std::map<uint32,server*>::iterator iter;
		iter = m_sendmap.find(socket);

		if(iter == m_sendmap.end())
		{
			m_log->WriteTrace("clear server error");
			m_lock.LeaveLock();
			return false;
		}
        
		m_sendlist.free_send(iter->second);
		m_sendmap.erase(iter);
		
		std::map<uint32,server_recv*>::iterator iter1;
		iter1 = m_recvmap.find(socket);

		if(iter1 == m_recvmap.end())
		{
			m_log->WriteTrace("clear server_recv error");
			m_lock.LeaveLock();
			return false;
		}

		m_recvlist.free_recv(iter1->second);
		m_recvmap.erase(iter1);

		m_server->del_socket(socket);

		char lls[1024] = {0};
		sprintf(lls,"down ok:socket=%d",socket);
		m_log->WriteTrace(lls);

		m_lock.LeaveLock();
		return true;
	}

	bool set_max_id(ip_node task,uint32 socket,uint32 last_id)
	{
		m_lock.EnterLock();

		ip_node_key node_key;
		node_key.d_ip = task.d_ip;
		node_key.d_port = task.d_port;
		node_key.l_port  = task.l_port;
		node_key.s_socket = socket;
				
		std::map<ip_node_key,uint32>::iterator it;
        
		it= m_map.find(node_key);
		if(it == m_map.end())
		{
			char ll[1024] = {0};
			sprintf(ll,"set_max_id:task && recv map error,socket=%d",socket);
			m_log->WriteTrace(ll);
         	m_lock.LeaveLock();
			return false;
		}

		if(it->second != socket)
		{
			m_log->WriteTrace("set_max_id:task:task && recv map error,socket error");
			m_lock.LeaveLock();
			return true;
		}

		std::map<uint32,server_recv*>::iterator iter2;
		iter2 = m_recvmap.find(socket);

		if(iter2 != m_recvmap.end())
		{
            uint32 rt = iter2->second->set_max_id(last_id);
			
			char ll[1024] = {0};
			sprintf(ll,"set_max_id lastid=%d,socket=%d",last_id,socket);
			m_log->WriteTrace(ll);
			
			if (rt ==2)
				OnError(socket);

			m_lock.LeaveLock();
			return true;
		}
		
		m_lock.LeaveLock();
		return false;
	}

private:

   tcp_server*  m_server;
   UDPServer*   m_udp;

   logfile*     m_log;
   lock         m_lock;

   ip_node      node[128];

   vector<ip_node>                             m_tasklist;

   std::map<ip_node,uint16>                    m_taskmap;
   std::map<ip_node_key,uint32>                m_map;
   
   std::map<uint32,server*>                    m_sendmap;
   std::map<uint32,server_recv*>               m_recvmap;
   
   send_map                                    m_sendlist;
   recv_map                                    m_recvlist;
};
#endif
